package com.sh.data.engine.job.admin.core.thread;

import com.sh.data.engine.job.admin.core.conf.XxlJobAdminConfig;
import com.sh.data.engine.job.admin.core.model.XxlJobLog;
import com.sh.data.engine.job.admin.core.trigger.TriggerTypeEnum;
import com.sh.data.engine.job.admin.core.trigger.XxlJobTrigger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * job trigger thread pool helper
 *
 * @author xuxueli 2018-07-03 21:08:07
 */
public abstract class JobTriggerPoolHelper {
    private static final Logger LOGGER = LoggerFactory.getLogger(JobTriggerPoolHelper.class);

    private static ThreadPoolExecutor fastTriggerPool;
    private static ThreadPoolExecutor slowTriggerPool;

    // job timeout count
    private static volatile long minTim = System.currentTimeMillis() / 60000;     // ms > min
    private static final ConcurrentMap<Integer, AtomicInteger> jobTimeoutCountMap = new ConcurrentHashMap<>();

    public static void start() {
        fastTriggerPool = new ThreadPoolExecutor(
            10,
            XxlJobAdminConfig.getAdminConfig().getTriggerPoolFastMax(),
            60L,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(1000),
            r -> new Thread(r, "xxl-job, admin JobTriggerPoolHelper-fastTriggerPool-" + r.hashCode()));

        slowTriggerPool = new ThreadPoolExecutor(
            10,
            XxlJobAdminConfig.getAdminConfig().getTriggerPoolSlowMax(),
            60L,
            TimeUnit.SECONDS,
            new LinkedBlockingQueue<>(2000),
            r -> new Thread(r, "xxl-job, admin JobTriggerPoolHelper-slowTriggerPool-" + r.hashCode()));
    }


    public static void stop() {
        //triggerPool.shutdown();
        fastTriggerPool.shutdownNow();
        slowTriggerPool.shutdownNow();
        LOGGER.info(">>>>>>>>> xxl-job trigger thread pool shutdown success.");
    }

    /**
     * @param jobId
     * @param triggerType
     * @param failRetryCount        >=0: use this param
     *                              <0: use param from job info config
     * @param executorShardingParam
     * @param executorParam         null: use job param
     *                              not null: cover job param
     */
    public static Future<List<XxlJobLog>> trigger(int jobId,
                                                  TriggerTypeEnum triggerType,
                                                  int failRetryCount,
                                                  String executorShardingParam,
                                                  String executorParam,
                                                  String addressList) {
        return addTrigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam, addressList);
    }

    private static Future<List<XxlJobLog>> addTrigger(int jobId,
                                                      TriggerTypeEnum triggerType,
                                                      int failRetryCount,
                                                      String executorShardingParam,
                                                      String executorParam,
                                                      String addressList) {

        // choose thread pool
        ThreadPoolExecutor triggerPool = fastTriggerPool;
        AtomicInteger jobTimeoutCount = jobTimeoutCountMap.get(jobId);
        if (jobTimeoutCount != null && jobTimeoutCount.get() > 10) {      // job-timeout 10 times in 1 min
            triggerPool = slowTriggerPool;
        }

        // trigger
        return triggerPool.submit(() -> {
            long start = System.currentTimeMillis();

            try {
                return XxlJobTrigger.trigger(jobId, triggerType, failRetryCount, executorShardingParam, executorParam, addressList);
            } catch (Exception e) {
                LOGGER.error(e.getMessage(), e);
                throw e;
            } finally {
                // check timeout-count-map
                long minTim_now = System.currentTimeMillis() / 60000;
                if (minTim != minTim_now) {
                    minTim = minTim_now;
                    jobTimeoutCountMap.clear();
                }

                // incr timeout-count-map
                long cost = System.currentTimeMillis() - start;
                if (cost > 500) {       // ob-timeout threshold 500ms
                    AtomicInteger timeoutCount = jobTimeoutCountMap.putIfAbsent(jobId, new AtomicInteger(1));
                    if (timeoutCount != null) {
                        timeoutCount.incrementAndGet();
                    }
                }
            }

        });
    }


}
